package streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object WordCount_SparkStreaming {
  def main(args: Array[String]): Unit = {
    //创建配置文件对象 注意：Streaming程序至少不能设置为local，至少需要2个线程
    val conf: SparkConf = new SparkConf().setAppName("Spark01_W").setMaster("local[*]")

    //创建Spark Streaming上下文环境对象
    val ssc = new StreamingContext(conf,Seconds(3))

    //操作数据源-从端口中获取一行数据
    val socketDS: ReceiverInputDStream[String] = ssc.socketTextStream("node01",9999)

    //对获取的一行数据进行扁平化操作
    val flatMapDS: DStream[String] = socketDS.flatMap(_.split(" "))

    //结构转换
    val mapDS: DStream[(String, Int)] = flatMapDS.map((_,1))

    //对数据进行聚合
    val reduceDS: DStream[(String, Int)] = mapDS.reduceByKey(_+_)

    //根据key对数据状态进行更新
    //传递的参数中含有两个值，第一个值表示相同key的value数据
    //第二个值表示缓存区相同key的value
    mapDS.updateStateByKey((seq:Seq[Int],buff:Option[Int]) => {
      val newCount = buff.getOrElse(0)+seq.sum
      Option(newCount)
    })

    //输出结果   注意：调用的是DS的print函数
    reduceDS.print()

    //启动采集器
    ssc.start()

    //默认情况下，上下文对象不能关闭
    //ssc.stop()

    //等待采集结束，终止上下文环境对象
    ssc.awaitTermination()
  }
}
